package spring.com.rocketmq.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;
import spring.com.rocketmq.config.RocketMqConsumerConfig;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;

/*
* rocketmq 消费者配置：订阅消息
 * */
@Slf4j
@Component //注入容器中
public class RocketMqConsumer {
    //引入消费者配置信息文件
    @Resource
    private RocketMqConsumerConfig config;
    //引入消费者实体对象
    private DefaultMQPushConsumer consumer;

    public RocketMqConsumer() throws MQClientException {
        //实例对象
        consumer = new DefaultMQPushConsumer("CicadaGroup");
        //地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //消费模式：顺序消费，并发消费 二种
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //主题和标签
        //如果要对应多个消费，那么每一个标签对应一个消费者，当然也可以将组进行分开为多个消费的，也行，具体，看业务
        //这里的  * 代表所有的tag标签,将 * 号换为单独的标签即可
        consumer.subscribe("DefaultCluster","*");
        //注册消费者的监听，并且在这个监听中进行消费信息
        //registerMessageListener

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs中只收集同一个topic，同一个tag，并且key相同的message
            // 会把不同的消息分别放置到不同的队列中
            try {
                for (Message msg : msgs) {
                    //消费者获取消息 这里只输出 不做后面逻辑处理
                    String body = new String(msg.getBody(), "utf-8");
                    log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                //如果出现消费失败：那么重试再次消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
            }
            //消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //注意：这里clientIP 我们无法修改，但是可以通过设置System的"rocketmq.client.name"属性值来生成不一样的clientId
        //每一个name我们可以设置为不同的组，就可以有不同的clientIP，每一个消费者都要定义一个最好
        System.setProperty("rocketmq.client.name", "CicadaGroup"); // 修改系统变量

        consumer.start();
        System.out.println("=========消费者 启动成功=======");



        //多个消费者：


        //实例对象
        consumer = new DefaultMQPushConsumer("CicadaGroup2");
        //地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //消费模式：顺序消费，并发消费 二种
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //主题和标签
        //如果要对应多个消费，那么每一个标签对应一个消费者，当然也可以将组进行分开为多个消费的，也行，具体，看业务
        //这里的  * 代表所有的tag标签,将 * 号换为单独的标签即可
        consumer.subscribe("DefaultCluster","*");
        //注册消费者的监听，并且在这个监听中进行消费信息
        //registerMessageListener

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs中只收集同一个topic，同一个tag，并且key相同的message
            // 会把不同的消息分别放置到不同的队列中
            try {
                for (Message msg : msgs) {
                    //消费者获取消息 这里只输出 不做后面逻辑处理
                    String body = new String(msg.getBody(), "utf-8");
                    log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                //如果出现消费失败：那么重试再次消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
            }
            //消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //注意：这里clientIP 我们无法修改，但是可以通过设置System的"rocketmq.client.name"属性值来生成不一样的clientId
        //每一个name我们可以设置为不同的组，就可以有不同的clientIP，每一个消费者都要定义一个最好
        System.setProperty("rocketmq.client.name", "CicadaGroup2"); // 修改系统变量

        consumer.start();
        System.out.println("=========消费者 启动成功=======");
    }
}
